#include	"SQL.h"
#include	<mysql.h>
#include	<cstring>
#include	<ctime>
#include	<Logger.h>
#include	<Utils.h>

#if defined(_WIN32) && defined(GetMessage)
#	undef	GetMessage
#	pragma	warning(disable: 4996)
#endif

namespace SQL {

	//////////////////////////////////////////////////////////////////////////
	/// Blob data encode
	//////////////////////////////////////////////////////////////////////////
	static string EncodeBlob(const char * data, size_t size) {
		static const char __hexstr[] = "0123456789ABCDEF";

		if (!data || size <= 0) return 0;

		string result;
		result.reserve(2 * size + 8);
		result.append("X'");

		for (size_t i = 0; i < size; ++i) {
			uint8_t c = (uint8_t)data[i];
			result.push_back(__hexstr[(c & 0xF0) >> 4]);
			result.push_back(__hexstr[c & 0x0F]);
		}

		result.push_back('\'');
		return move(result);
	}

	//////////////////////////////////////////////////////////////////////////
	/// Helper class for read/write data into buffer.
	//////////////////////////////////////////////////////////////////////////
	class Reader {
	public:
		Reader(const char * data) : _data(data), _offset(0) {}

		bool		Bool() { bool * p = (bool *)(_data + _offset); _offset += sizeof(bool); return *p; }
		int32_t		Int32() { int32_t * p = (int32_t *)(_data + _offset); _offset += sizeof(int32_t); return *p; }
		int64_t		Int64() { int64_t * p = (int64_t *)(_data + _offset); _offset += sizeof(int64_t); return *p; }
		uint32_t	UInt32() { uint32_t * p = (uint32_t *)(_data + _offset); _offset += sizeof(uint32_t); return *p; }
		uint64_t	UInt64() { uint64_t * p = (uint64_t *)(_data + _offset); _offset += sizeof(uint64_t); return *p; }
		double		Double() { double * p = (double *)(_data + _offset); _offset += sizeof(double); return *p; }
		float		Float() { float * p = (float *)(_data + _offset); _offset += sizeof(float); return *p; }
		string		Data() {
			uint16_t * size = (uint16_t *)(_data + _offset);
			char * src = (char *)(_data + _offset + sizeof(uint16_t));
			_offset += sizeof(uint16_t) + *size;
			return string(src, *size);
		}
		
	private:
		const char * _data;
		size_t _offset;
	};

	class Writer {
	public:
		Writer(char * buffer) : _buffer(buffer), _offset(0) {}

		size_t Size() { return _offset; }

		void Bool(bool v) { bool * p = (bool *)(_buffer + _offset); _offset += sizeof(bool); *p = v; }
		void Int32(int32_t v) { int32_t * p = (int32_t *)(_buffer + _offset); _offset += sizeof(int32_t); *p = v; }
		void Int64(int64_t v) { int64_t * p = (int64_t *)(_buffer + _offset); _offset += sizeof(int64_t); *p = v; }
		void UInt32(uint32_t v) { uint32_t * p = (uint32_t *)(_buffer + _offset); _offset += sizeof(uint32_t); *p = v; }
		void UInt64(uint64_t v) { uint64_t * p = (uint64_t *)(_buffer + _offset); _offset += sizeof(uint64_t); *p = v; }
		void Double(double v) { double * p = (double *)(_buffer + _offset); _offset += sizeof(double); *p = v; }
		void Float(float v) { float * p = (float *)(_buffer + _offset); _offset += sizeof(float); *p = v; }
		void Data(const string & v) {
			uint16_t * len = (uint16_t *)(_buffer + _offset);
			char * dst = _buffer + _offset + sizeof(uint16_t);
			*len = (uint16_t)v.size();
			_offset += sizeof(uint16_t) + v.size();
			memcpy(dst, v.data(), v.size());
		}
		void Data(const google::protobuf::Message & data) {
			uint16_t * len = (uint16_t *)(_buffer + _offset);
			char * dst = _buffer + _offset + sizeof(uint16_t);
			size_t size = data.ByteSizeLong();
			*len = (uint16_t)size;
			_offset += sizeof(uint16_t) + size;
			if (!data.SerializeToArray(dst, (int)size)) LOG_ERR("[SQL]Write protobuf to blob failed! Type : %s", data.GetTypeName().c_str());
		}

	private:
		char * _buffer;
		size_t _offset;
	};

	//////////////////////////////////////////////////////////////////////////
	/// SQL::Database
	//////////////////////////////////////////////////////////////////////////
	bool Database::Open(const string & host, int port, const string & user, const string & pswd, const string & db) {
		if (_valid) Close();

		_ctx = new MYSQL();
		char reconnect = 1;
		
		mysql_init(_ctx);
		mysql_options(_ctx, MYSQL_OPT_RECONNECT, &reconnect);

		if (!mysql_real_connect(_ctx, host.data(), user.data(), pswd.data(), db.data(), port, 0, 0)) {
			LOG_ERR("[SQL]real_connect : %s", mysql_error(_ctx));
			delete _ctx;
			return false;
		}

		_valid = true;
		return true;
	}

	void Database::Close() {
		if (!_valid) return;
		mysql_close(_ctx);
		delete _ctx;
		_valid = false;
	}

	int Database::Exec(const string & sql) {
		if (!_valid) {
			LOG_ERR("[SQL]Try to exec sql on invalid database : %s", sql.c_str());
			return 0;
		}

		if (mysql_real_query(_ctx, sql.data(), (unsigned long)sql.size()) != 0) {
			LOG_ERR("[SQL]Error: %s. SQL: %s", mysql_error(_ctx), sql.c_str());
			return 0;
		}

		LOG_DEBUG("[SQL]%s", sql.c_str());
		return (int)mysql_affected_rows(_ctx);
	}

	SharedCursor Database::Query(const string & sql) {
		if (!_valid) {
			LOG_ERR("[SQL]Try to exec sql on invalid database : %s", sql.c_str());
			return make_shared<Cursor>(nullptr);
		}

		if (mysql_real_query(_ctx, sql.data(), (unsigned long)sql.size()) != 0) {
			LOG_ERR("[SQL]Error: %s. SQL: %s", mysql_error(_ctx), sql.c_str());
			return make_shared<Cursor>(nullptr);
		}

		LOG_DEBUG("[SQL]%s", sql.c_str());
		return make_shared<Cursor>(mysql_store_result(_ctx));
	}

	string Database::Escape(const string & source) {
		char * buffer = new char[source.size() * 2];
		size_t size = mysql_real_escape_string(_ctx, buffer, source.data(), (unsigned long)source.size());
		string result(buffer, size);
		delete[] buffer;
		return move(result);
	}

	int64_t Database::LastInsertId() {
		return (int64_t)mysql_insert_id(_ctx);
	}

	int Database::Insert(const string & table, const google::protobuf::Message & data) {
		auto reflect = data.GetReflection();
		auto desc = data.GetDescriptor();
		int fields = desc->field_count();

		string keys, values;
		keys.reserve(1024);
		values.reserve(65536);

		for (int i = 0; i < fields; ++i) {
			auto field = desc->field(i);
			if (field->is_extension()) continue;
			if (field->is_repeated() && reflect->FieldSize(data, field) <= 0) continue;
			if (!field->is_repeated() && !reflect->HasField(data, field)) continue;

			keys.push_back('`');
			keys.append(field->lowercase_name());
			keys.append("`,");

			values.append(__Serialize(data, reflect, field));
			values.push_back(',');
		}

		keys.pop_back();
		values.pop_back();

		string sql;
		sql.reserve(keys.size() + values.size() + 256);
		sql.append("INSERT INTO `");
		sql.append(ToLower(table));
		sql.append("`(");
		sql.append(keys);
		sql.append(") VALUES(");
		sql.append(values);
		sql.push_back(')');

		return Exec(sql);
	}

	int Database::Update(const string & table, const google::protobuf::Message & data, const string & key) {
		auto reflect = data.GetReflection();
		auto desc = data.GetDescriptor();
		int fields = desc->field_count();

		std::string sql, where = " WHERE " + key + "=";
		sql.reserve(131072);
		sql.append("UPDATE `");
		sql.append(ToLower(table));
		sql.append("` SET ");

		for (int i = 0; i < fields; ++i) {
			auto field = desc->field(i);
			if (field->is_extension()) continue;
			if (field->is_repeated() && reflect->FieldSize(data, field) <= 0) continue;
			if (!field->is_repeated() && !reflect->HasField(data, field)) continue;

			string name = field->lowercase_name();
			string val = __Serialize(data, reflect, field);

			if (name == key) where.append(val);

			sql.append(name);
			sql.push_back('=');
			sql.append(val);
			sql.push_back(',');
		}

		sql.pop_back();
		sql.append(where);
		return Exec(sql);
	}

	string Database::__Serialize(const google::protobuf::Message & source, const google::protobuf::Reflection * reflect, const google::protobuf::FieldDescriptor * field) {
		if (field->is_repeated()) {
			char * buffer = new char[131072];
			int count = reflect->FieldSize(source, field);

			Writer w(buffer);
			w.Int32((int32_t)count);

			switch (field->cpp_type()) {
			case google::protobuf::FieldDescriptor::CPPTYPE_INT32:
				for (int i = 0; i < count; ++i) w.Int32(reflect->GetRepeatedInt32(source, field, i));
				break;
			case google::protobuf::FieldDescriptor::CPPTYPE_INT64:
				for (int i = 0; i < count; ++i) w.Int64(reflect->GetRepeatedInt64(source, field, i));
				break;
			case google::protobuf::FieldDescriptor::CPPTYPE_UINT32:
				for (int i = 0; i < count; ++i) w.UInt32(reflect->GetRepeatedUInt32(source, field, i));
				break;
			case google::protobuf::FieldDescriptor::CPPTYPE_UINT64:
				for (int i = 0; i < count; ++i) w.UInt64(reflect->GetRepeatedUInt64(source, field, i));
				break;
			case google::protobuf::FieldDescriptor::CPPTYPE_DOUBLE:
				for (int i = 0; i < count; ++i) w.Double(reflect->GetRepeatedDouble(source, field, i));
				break;
			case google::protobuf::FieldDescriptor::CPPTYPE_FLOAT:
				for (int i = 0; i < count; ++i) w.Float(reflect->GetRepeatedFloat(source, field, i));
				break;
			case google::protobuf::FieldDescriptor::CPPTYPE_BOOL:
				for (int i = 0; i < count; ++i) w.Bool(reflect->GetRepeatedBool(source, field, i));
				break;
			case google::protobuf::FieldDescriptor::CPPTYPE_ENUM:
				for (int i = 0; i < count; ++i) w.Int32(reflect->GetRepeatedEnumValue(source, field, i));
				break;
			case google::protobuf::FieldDescriptor::CPPTYPE_STRING:
				for (int i = 0; i < count; ++i) w.Data(reflect->GetRepeatedString(source, field, i));
				break;
			case google::protobuf::FieldDescriptor::CPPTYPE_MESSAGE:
				for (int i = 0; i < count; ++i) w.Data(reflect->GetRepeatedMessage(source, field, i));
				break;
			}

			string result = EncodeBlob(buffer, w.Size());
			delete[] buffer;
			return move(result);
		} else {
			switch (field->cpp_type()) {
			case google::protobuf::FieldDescriptor::CPPTYPE_INT32: return to_string(reflect->GetInt32(source, field));
			case google::protobuf::FieldDescriptor::CPPTYPE_INT64: return to_string(reflect->GetInt64(source, field));
			case google::protobuf::FieldDescriptor::CPPTYPE_UINT32: return to_string(reflect->GetUInt32(source, field));
			case google::protobuf::FieldDescriptor::CPPTYPE_UINT64: return to_string(reflect->GetUInt64(source, field));
			case google::protobuf::FieldDescriptor::CPPTYPE_DOUBLE: return to_string(reflect->GetDouble(source, field));
			case google::protobuf::FieldDescriptor::CPPTYPE_FLOAT: return to_string(reflect->GetFloat(source, field));
			case google::protobuf::FieldDescriptor::CPPTYPE_BOOL: return reflect->GetBool(source, field) ? "1" : "0";
			case google::protobuf::FieldDescriptor::CPPTYPE_ENUM: return to_string(reflect->GetEnumValue(source, field));
			case google::protobuf::FieldDescriptor::CPPTYPE_STRING: return "'" + Escape(reflect->GetString(source, field)) + "'";
			case google::protobuf::FieldDescriptor::CPPTYPE_MESSAGE: {
				auto & data = reflect->GetMessage(source, field);
				size_t size = data.ByteSizeLong();
				char * buffer = new char[size];

				if (!data.SerializeToArray(buffer, (int)size)) {
					LOG_ERR("[SQL]Serialize message failed! Type : %s", data.GetTypeName().c_str());
					delete[] buffer;
					return "";
				}

				string result = EncodeBlob(buffer, size);
				delete[] buffer;
				return result; }
			}

			return "";
		}
	}

	//////////////////////////////////////////////////////////////////////////
	/// SQL::Cursor
	//////////////////////////////////////////////////////////////////////////
	Cursor::Cursor(st_mysql_res * res) : _res(res), _current(nullptr) {
		if (!_res) return;

		int fields = mysql_num_fields(_res);
		if (fields <= 0) return;

		for (int n = 0; n < fields; ++n) {
			MYSQL_FIELD * field = mysql_fetch_field(_res);
			_key2idx[field->name] = n;
			_fields.push_back(field);
		}

		mysql_data_seek(_res, 0);
	}

	Cursor::~Cursor() {
		if (!_res) return;
		while (mysql_fetch_row(_res)) continue;
		mysql_free_result(_res);
	}

	bool Cursor::Next() {
		if (!_res) return false;

		_current = mysql_fetch_row(_res);
		_len = mysql_fetch_lengths(_res);
		return _current != nullptr && _len != nullptr;
	}

	bool Cursor::IsNull(const string & column) {
		auto it = _key2idx.find(column);
		if (it == _key2idx.end()) return true;
		return _current[it->second] == nullptr || _len[it->second] == 0;
	}

	template<>
	bool Cursor::Get<bool>(const string & column) {
		int n = __Check4Get(column);
		if (n < 0 || !_current[n]) return false;
		return _current[n] == "1" || _current[n] == "true";
	}

	template<>
	int8_t Cursor::Get<int8_t>(const string & column) {
		int n = __Check4Get(column);
		if (n < 0 || !_current[n]) return 0;
		return (int8_t)atoi(_current[n]);
	}

	template<>
	uint8_t Cursor::Get<uint8_t>(const string & column) {
		int n = __Check4Get(column);
		if (n < 0 || !_current[n]) return 0;
		return (uint8_t)atoi(_current[n]);
	}

	template<>
	int16_t Cursor::Get<int16_t>(const string & column) {
		int n = __Check4Get(column);
		if (n < 0 || !_current[n]) return 0;
		return (int16_t)atoi(_current[n]);
	}

	template<>
	uint16_t Cursor::Get<uint16_t>(const string & column) {
		int n = __Check4Get(column);
		if (n < 0 || !_current[n]) return 0;
		return (uint16_t)atoi(_current[n]);
	}

	template<>
	int32_t Cursor::Get<int32_t>(const string & column) {
		int n = __Check4Get(column);
		if (n < 0 || !_current[n]) return 0;
		return (int32_t)atoi(_current[n]);
	}

	template<>
	uint32_t Cursor::Get<uint32_t>(const string & column) {
		int n = __Check4Get(column);
		if (n < 0 || !_current[n]) return 0;
		return (uint32_t)atoi(_current[n]);
	}

	template<>
	int64_t Cursor::Get<int64_t>(const string & column) {
		int n = __Check4Get(column);
		if (n < 0 || !_current[n]) return 0;
		return (int64_t)atoll(_current[n]);
	}

	template<>
	uint64_t Cursor::Get<uint64_t>(const string & column) {
		int n = __Check4Get(column);
		if (n < 0 || !_current[n]) return 0;
		return (uint64_t)atoll(_current[n]);
	}

	template<>
	float Cursor::Get<float>(const string & column) {
		int n = __Check4Get(column);
		if (n < 0 || !_current[n]) return 0;
		return (float)atof(_current[n]);
	}

	template<>
	double Cursor::Get<double>(const string & column) {
		int n = __Check4Get(column);
		if (n < 0 || !_current[n]) return 0;
		return atof(_current[n]);
	}

	template<>
	string Cursor::Get<string>(const string & column) {
		int n = __Check4Get(column);
		if (n < 0 || !_current[n]) return "";
		return string(_current[n], _len[n]);
	}

	time_t Cursor::GetTimestamp(const string & column) {
		int nY = 1970, nMonth = 1, nD = 1, nH = 0, nM = 0, nS = 0;
		int n = __Check4Get(column);
		if (n >= 0) sscanf(_current[n], "%d-%d-%d %d:%d:%d", &nY, &nMonth, &nD, &nH, &nM, &nS);
		struct tm iInfo;
		iInfo.tm_year = nY - 1900;
		iInfo.tm_mon = nMonth - 1;
		iInfo.tm_mday = nD;
		iInfo.tm_hour = nH;
		iInfo.tm_min = nM;
		iInfo.tm_sec = nS;
		return mktime(&iInfo);
	}

	int Cursor::__Check4Get(const string & column) {
		auto it = _key2idx.find(column);
		if (it == _key2idx.end()) return -1;
		return it->second;
	}

	void Cursor::__Cast(google::protobuf::Message & msg) {
		auto data = &msg;
		auto reflect = msg.GetReflection();
		auto desc = msg.GetDescriptor();
		int fields = desc->field_count();

		for (int i = 0; i < fields; ++i) {
			auto field = desc->field(i);
			auto name = field->lowercase_name();
			if (IsNull(name)) continue;
			
			if (field->is_repeated()) {
				string blob = Get<string>(name);
				Reader r(blob.data());

				int count = r.Int32();
				switch (field->cpp_type()) {
				case google::protobuf::FieldDescriptor::CPPTYPE_INT32:
					for (int i = 0; i < count; ++i) reflect->AddInt32(data, field, r.Int32());
					break;
				case google::protobuf::FieldDescriptor::CPPTYPE_INT64:
					for (int i = 0; i < count; ++i) reflect->AddInt64(data, field, r.Int64());
					break;
				case google::protobuf::FieldDescriptor::CPPTYPE_UINT32:
					for (int i = 0; i < count; ++i) reflect->AddUInt32(data, field, r.UInt32());
					break;
				case google::protobuf::FieldDescriptor::CPPTYPE_UINT64:
					for (int i = 0; i < count; ++i) reflect->AddUInt64(data, field, r.UInt64());
					break;
				case google::protobuf::FieldDescriptor::CPPTYPE_DOUBLE:
					for (int i = 0; i < count; ++i) reflect->AddDouble(data, field, r.Double());
					break;
				case google::protobuf::FieldDescriptor::CPPTYPE_FLOAT:
					for (int i = 0; i < count; ++i) reflect->AddFloat(data, field, r.Float());
					break;
				case google::protobuf::FieldDescriptor::CPPTYPE_BOOL:
					for (int i = 0; i < count; ++i) reflect->AddBool(data, field, r.Bool());
					break;
				case google::protobuf::FieldDescriptor::CPPTYPE_ENUM:
					for (int i = 0; i < count; ++i) reflect->AddEnumValue(data, field, r.Int32());
					break;
				case google::protobuf::FieldDescriptor::CPPTYPE_STRING:
					for (int i = 0; i < count; ++i) reflect->AddString(data, field, r.Data());
					break;
				case google::protobuf::FieldDescriptor::CPPTYPE_MESSAGE:
					for (int i = 0; i < count; ++i) {
						string sub = r.Data();
						if (!reflect->AddMessage(data, field)->ParseFromArray(sub.data(), (int)sub.size())) {
							LOG_ERR("[SQL]Cast failed. Field : %s. Type : %s", name.c_str(), desc->name().c_str());
						}
					}
					break;
				}
			} else {
				switch (field->cpp_type()) {
				case google::protobuf::FieldDescriptor::CPPTYPE_INT32: reflect->SetInt32(data, field, Get<int32_t>(name)); break;
				case google::protobuf::FieldDescriptor::CPPTYPE_INT64: reflect->SetInt64(data, field, Get<int64_t>(name)); break;
				case google::protobuf::FieldDescriptor::CPPTYPE_UINT32: reflect->SetUInt32(data, field, Get<uint32_t>(name)); break;
				case google::protobuf::FieldDescriptor::CPPTYPE_UINT64: reflect->SetUInt64(data, field, Get<uint64_t>(name)); break;
				case google::protobuf::FieldDescriptor::CPPTYPE_DOUBLE: reflect->SetDouble(data, field, Get<double>(name)); break;
				case google::protobuf::FieldDescriptor::CPPTYPE_FLOAT: reflect->SetFloat(data, field, Get<float>(name));; break;
				case google::protobuf::FieldDescriptor::CPPTYPE_BOOL: reflect->SetBool(data, field, Get<bool>(name)); break;
				case google::protobuf::FieldDescriptor::CPPTYPE_ENUM: reflect->SetEnumValue(data, field, Get<int32_t>(name)); break;
				case google::protobuf::FieldDescriptor::CPPTYPE_STRING: reflect->SetString(data, field, Get<string>(name)); break;
				case google::protobuf::FieldDescriptor::CPPTYPE_MESSAGE: {
					string blob = Get<string>(name);
					if (!reflect->MutableMessage(data, field)->ParseFromArray(blob.data(), (int)blob.size())) {
						LOG_ERR("[SQL]Cast failed. Field : %s. Type : %s", name.c_str(), desc->name().c_str());
					}
					break; }
				}
			}
		}
	}

	//////////////////////////////////////////////////////////////////////////
	/// SQL::Pool
	//////////////////////////////////////////////////////////////////////////
	Pool::Pool(int workers) : _valid(false), _size(workers), _jobs(), _workers(), _lock(), _signal() {
	}

	Pool::~Pool() {
		_valid = false;
		_signal.notify_all();
		while (!_jobs.empty()) this_thread::yield();

		while (!_workers.empty()) {
			thread * p = _workers.front();
			_workers.pop();
			if (p->joinable()) p->join();
			delete p;
		}
	}

	bool Pool::Open(const string & host, int port, const string & user, const string & pswd, const string & db) {
		_valid = true;

		for (int i = 0; i < _size; ++i) {
			try {
				Database * conn = new Database();
				if (!conn->Open(host, port, user, pswd, db)) {
					_valid = false;
					break;
				}

				_workers.push(new thread(&Pool::__Work, this, conn));
			} catch (runtime_error & e) {
				LOG_ERR("[SQL]Failed to connect to %s:%d due to : %s", host.c_str(), port, e.what());
				_valid = false;
				break;
			}
		}

		return _valid;
	}

	void Pool::__Work(Database * db) {
		while (true) {
			unique_lock<mutex> lock(_lock);
			if (_jobs.empty()) {
				if (!_valid) break;
				_signal.wait(lock);
				if (_jobs.empty()) continue;
			}

			ITask * p = _jobs.front();
			_jobs.pop();
			lock.unlock();

			p->OnRun(*db);
			delete p;
		}

		delete db;
	}
}